package com.gdlt.mq2db;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * Created by CM on 2017/3/14.
 */
public class PGWriter {
    Connection connection;
    String topic;
    int partitionId;

    private String recValue;
    private String ip;
    private String index;
    private BigDecimal value;// 对应数据库的decimal类型，若用double精度显示不全
    private int ts;

    public Connection getConn() { // 得到数据库的连接

        Configuration conf = Kafka2pgConfig.getConf();
        String url = conf.get(Kafka2pgConfig.PG_URL_CONF_NAME);
        String user = conf.get(Kafka2pgConfig.PG_USER_CONF_NAME);
        String password = conf.get(Kafka2pgConfig.PG_PASSWORD_CONF_NAME);
        String driver = conf.get(Kafka2pgConfig.PG_DRIVER_CONF_NAME);
        Connection connection = null;// 数据库连接
        try {
            Class.forName(driver);// 加载驱动程序，你的jar包要用上啊
            connection = DriverManager.getConnection(url, user, password);
            System.out.println("成功连接pg数据库:" + connection);
        } catch (ClassNotFoundException e) {
            // LOG.error("未加载数据库驱动程序" + e);
        } catch (SQLException e) {
            // LOG.error("连接数据库异常" + e);
        }
        return connection;
    }

    // public PGWriter(String topic,int partitionId){
    //
    // }

    /**
     * get offset from db
     * @return
     */
    public long getLastoffset() {
        long lastOffset = -1;
        connection = getConn();
        PreparedStatement pstmt = null;
        String sql = "select max(recset) from golia_offset";

        try {
            pstmt = connection.prepareStatement(sql);
            ResultSet rs = pstmt.executeQuery();
            if (rs.next()) {
                lastOffset = rs.getLong(1);
            }

        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return lastOffset;
    }


    /**
     * write data and offset
     * @param data
     * @param offset
     */

    public void write(List<ConsumerRecord<String, String>> data, long offset) {
        connection = getConn();// 获得数据库连接
        PreparedStatement pstmt = null; // 数据库操作，插入数据到golia_data表中
        PreparedStatement pstmtSet = null; // 数据库操作，插入数据到golia_offset表中
        String sql = "insert into golia_data(ip,index,value,ts) select ?,?,?,?"
                + "where not exists(select ip,index,ts from golia_data where " + "ip = ? and index = ? and ts = ?)";
        String sqlSet = "insert into golia_offset(recset,partition) select ?,?"
                + "where not exists(select recset,partition from golia_offset where" + "recset = ? and partition = ?)";

        try {
            connection.setAutoCommit(false);
            pstmt = connection.prepareStatement(sql);
            pstmtSet = connection.prepareStatement(sqlSet);

            for (int i = 0; i < data.size(); i++) { // 先对消息队列中的value值根据空格划分
                recValue = data.get(i).value();
                String splitBlank[] = recValue.split(" ");
                String splitComma[] = splitBlank[0].split("\\.");// 根据点号分隔

                ip = splitComma[1].replace('_', '.');
                index = splitBlank[0].substring(splitComma[0].length() + splitComma[1].length() + 2);
                value = new BigDecimal(splitBlank[1]);// 传入String
                ts = Integer.parseInt(splitBlank[2]);

                pstmt.setString(1, ip);
                pstmt.setString(2, index);
                pstmt.setBigDecimal(3, value);
                pstmt.setInt(4, ts);
                pstmt.setString(5, ip);
                pstmt.setString(6, index);
                pstmt.setInt(7, ts);
                pstmt.addBatch();

            }
            int[] counts = pstmt.executeBatch();
            // connection.setAutoCommit(true);

            offset = data.get(data.size() - 1).offset();
            pstmtSet.setLong(1, offset);
            pstmtSet.setInt(2, partitionId);
            pstmtSet.setLong(3, offset);
            pstmtSet.setInt(4, partitionId);
            pstmtSet.executeUpdate(sqlSet);

            connection.commit();// 一起提交,这个可以跟offset表的一起提交么

            int dupli = 0, fails = 0;
            for (int i = 0; i < counts.length; i++) {
                if (counts[i] == 0) {
                    dupli++;
                }

                if (counts[i] == Statement.EXECUTE_FAILED) {
                    fails++;
                }
            }
            int inPg = counts.length - dupli - fails;
            System.out.println("这批数据共" + counts.length + "条，插入数据库有" +
                    inPg + "条，其中有" + dupli + "条记录键值重复，" + fails + "条记录执行失败。");

            pstmt.close(); // 操作关闭
            pstmtSet.close(); // 操作关闭

        } catch (SQLException e) {
            try {
                connection.rollback();// 回滚操作
                // LOG.info("回滚成功");
            } catch (Exception e2) {
                // TODO: handle exception
                // LOG.error("回滚异常" + e2);
            }
            // TODO Auto-generated catch block
            // LOG.error("插入数据库异常:" + e);
        }

    }

    public void close() {
        try {
            connection.close();
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}
